package com.bins.cloud.stream.mq;

import com.bins.cloud.stream.eo.mq.MqMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

@Configuration
@Slf4j
public class RocketMQComprehensive {

    @Bean
    public Supplier<Flux<MqMsg>> functionProducer() {
        return () -> Flux.interval(Duration.ofSeconds(2)).map(id -> {
            MqMsg mqMsg = new MqMsg();
            mqMsg.setId(id.toString());
            mqMsg.setName("freeman");
            return mqMsg;
        }).log();
    }

    @Bean
    public Function<Flux<MqMsg>, Flux<MqMsg>> functionProcessor() {
        return flux -> flux.map(mqMsg -> {
            log.info("MQ信息：{}" ,mqMsg.toString());
            mqMsg.setId(String.valueOf(Long.parseLong(mqMsg.getId()) * Long.parseLong(mqMsg.getId())));
            return mqMsg;
        });
    }

	@Bean
	public Consumer<MqMsg> functionConsumer() {
		return num -> {
			log.info("接收到消息："+ num.toString());
		};
	}




}